/*-------------------------------------------------------------------------
 *
 * nodeSort.c
 *	  处理关系排序的例程。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/executor/nodeSort.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/parallel.h"
#include "executor/execdebug.h"
#include "executor/nodeSort.h"
#include "miscadmin.h"
#include "utils/tuplesort.h"


/* ----------------------------------------------------------------
 *		ExecSort
 *
 *		使用tuplesort对节点的外部子树中的元组进行排序，
 *		将结果保存在临时文件或内存中。在初始调用后，
 *		每次调用从文件中返回一个元组。
 *
 *		此排序可以通过两种不同的方式执行：
 *
 *		1) 当结果是单列时，我们执行Datum排序。
 *
 *		2) 当结果包含多列时，我们执行元组排序。
 *
 *		我们可以通过始终执行元组排序来完成此任务，但是排序
 *		Datum的速度可能显著快于排序元组，
 *		尤其是当Datum为按值传递类型时。
 *
 *		条件：
 *		  -- 无。
 *
 *		初始状态：
 *		  -- 外部子节点准备返回第一个元组。
 * ----------------------------------------------------------------
 */
static TupleTableSlot * fc_ExecSort(PlanState *fc_pstate)
{
	SortState  *fc_node = castNode(SortState, fc_pstate);
	EState	   *fc_estate;
	ScanDirection fc_dir;
	Tuplesortstate *fc_tuplesortstate;
	TupleTableSlot *fc_slot;

	CHECK_FOR_INTERRUPTS();

	/*
	 * 从节点获取状态信息
	 */
	SO1_printf("ExecSort: %s\n",
			   "entering routine");

	fc_estate = fc_node->ss.ps.state;
	fc_dir = fc_estate->es_direction;
	fc_tuplesortstate = (Tuplesortstate *) fc_node->tuplesortstate;

	/*
	 * 如果是第一次循环，从外部计划中读取所有元组并将其传递给
	 * tuplesort.c。后续调用只是从tuplesort中获取元组。
	 */

	if (!fc_node->sort_Done)
	{
		Sort	   *fc_plannode = (Sort *) fc_node->ss.ps.plan;
		PlanState  *fc_outerNode;
		TupleDesc	fc_tupDesc;
		int			fc_tuplesortopts = TUPLESORT_NONE;

		SO1_printf("ExecSort: %s\n",
				   "sorting subplan");

		/*
		 * 想在创建排序数据时按前进方向扫描子计划。
		 */
		fc_estate->es_direction = ForwardScanDirection;

		/*
		 * 初始化tuplesort模块。
		 */
		SO1_printf("ExecSort: %s\n",
				   "calling tuplesort_begin");

		fc_outerNode = outerPlanState(fc_node);
		fc_tupDesc = ExecGetResultType(fc_outerNode);

		if (fc_node->randomAccess)
			fc_tuplesortopts |= TUPLESORT_RANDOMACCESS;
		if (fc_node->bounded)
			fc_tuplesortopts |= TUPLESORT_ALLOWBOUNDED;

		if (fc_node->datumSort)
			fc_tuplesortstate = tuplesort_begin_datum(TupleDescAttr(fc_tupDesc, 0)->atttypid,
												   fc_plannode->sortOperators[0],
												   fc_plannode->collations[0],
												   fc_plannode->nullsFirst[0],
												   work_mem,
												   NULL,
												   fc_tuplesortopts);
		else
			fc_tuplesortstate = tuplesort_begin_heap(fc_tupDesc,
												  fc_plannode->numCols,
												  fc_plannode->sortColIdx,
												  fc_plannode->sortOperators,
												  fc_plannode->collations,
												  fc_plannode->nullsFirst,
												  work_mem,
												  NULL,
												  fc_tuplesortopts);
		if (fc_node->bounded)
			tuplesort_set_bound(fc_tuplesortstate, fc_node->bound);
		fc_node->tuplesortstate = (void *) fc_tuplesortstate;

		/*
		 * 扫描子计划并使用基于我们正在执行的排序类型
		 * 的适当方法将所有元组传递给tuplesort。
		 */
		if (fc_node->datumSort)
		{
			for (;;)
			{
				fc_slot = ExecProcNode(fc_outerNode);

				if (TupIsNull(fc_slot))
					break;
				slot_getsomeattrs(fc_slot, 1);
				tuplesort_putdatum(fc_tuplesortstate,
								   fc_slot->tts_values[0],
								   fc_slot->tts_isnull[0]);
			}
		}
		else
		{
			for (;;)
			{
				fc_slot = ExecProcNode(fc_outerNode);

				if (TupIsNull(fc_slot))
					break;
				tuplesort_puttupleslot(fc_tuplesortstate, fc_slot);
			}
		}

		/*
		 * 完成排序。
		 */
		tuplesort_performsort(fc_tuplesortstate);

		/*
		 * 恢复到用户指定的方向
		 */
		fc_estate->es_direction = fc_dir;

		/*
		 * 最终将排序标志设置为true
		 */
		fc_node->sort_Done = true;
		fc_node->bounded_Done = fc_node->bounded;
		fc_node->bound_Done = fc_node->bound;
		if (fc_node->shared_info && fc_node->am_worker)
		{
			TuplesortInstrumentation *fc_si;

			Assert(IsParallelWorker());
			Assert(ParallelWorkerNumber <= fc_node->shared_info->num_workers);
			fc_si = &fc_node->shared_info->sinstrument[ParallelWorkerNumber];
			tuplesort_get_stats(fc_tuplesortstate, fc_si);
		}
		SO1_printf("ExecSort: %s\n", "sorting done");
	}

	SO1_printf("ExecSort: %s\n",
			   "retrieving tuple from tuplesort");

	fc_slot = fc_node->ss.ps.ps_ResultTupleSlot;

	/*
	 * 从适当的tuplesort函数中获取下一个排序项。对于
	 * datum排序，我们必须自己管理槽，并在
	 * tuplesort_getdatum返回false时将其清空以指示没有更多datum。
	 * 对于元组排序，tuplesort_gettupleslot为我们管理槽，并
	 * 在耗尽元组时清空槽。
	 */
	if (fc_node->datumSort)
	{
		ExecClearTuple(fc_slot);
		if (tuplesort_getdatum(fc_tuplesortstate, ScanDirectionIsForward(fc_dir),
							   &(fc_slot->tts_values[0]), &(fc_slot->tts_isnull[0]), NULL))
			ExecStoreVirtualTuple(fc_slot);
	}
	else
		(void) tuplesort_gettupleslot(fc_tuplesortstate,
									  ScanDirectionIsForward(fc_dir),
									  false, fc_slot, NULL);

	return fc_slot;
}

/* ----------------------------------------------------------------
 *		ExecInitSort
 *
 *		创建由规划器生成的排序节点的运行时状态信息
 *		并初始化其外部子树。
 * ----------------------------------------------------------------
 */
SortState *
ExecInitSort(Sort *fc_node, EState *fc_estate, int fc_eflags)
{
	SortState  *fc_sortstate;
	TupleDesc	fc_outerTupDesc;

	SO1_printf("ExecInitSort: %s\n",
			   "initializing sort node");

	/*
	 * 创建状态结构
	 */
	fc_sortstate = makeNode(SortState);
	fc_sortstate->ss.ps.plan = (Plan *) fc_node;
	fc_sortstate->ss.ps.state = fc_estate;
	fc_sortstate->ss.ps.ExecProcNode = fc_ExecSort;

	/*
	 * 我们必须随机访问排序输出才能进行向后扫描或
	 * 标记/恢复。如果我们可能会被调用多次回放，
	 * 我们还希望对排序输出进行物化。
	 */
	fc_sortstate->randomAccess = (fc_eflags & (EXEC_FLAG_REWIND |
										 EXEC_FLAG_BACKWARD |
										 EXEC_FLAG_MARK)) != 0;

	fc_sortstate->bounded = false;
	fc_sortstate->sort_Done = false;
	fc_sortstate->tuplesortstate = NULL;

	/*
	 * 杂项初始化
	 *
	 * 排序节点不初始化它们的 ExprContexts，因为它们从不调用 ExecQual 或 ExecProject。
	 */

	/*
	 * 初始化子节点
	 *
	 * 我们将子节点与支持 REWIND、BACKWARD 或
	 * MARK/RESTORE 的需求隔离。
	 */
	fc_eflags &= ~(EXEC_FLAG_REWIND | EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK);

	outerPlanState(fc_sortstate) = ExecInitNode(outerPlan(fc_node), fc_estate, fc_eflags);

	/*
	 * 初始化扫描槽和类型。
	 */
	ExecCreateScanSlotFromOuterPlan(fc_estate, &fc_sortstate->ss, &TTSOpsVirtual);

	/* 
	 * 初始化返回插槽和类型。无需初始化投影信息
	 * 因为此节点不执行投影。
	 */
	ExecInitResultTupleSlotTL(&fc_sortstate->ss.ps, &TTSOpsMinimalTuple);
	fc_sortstate->ss.ps.ps_ProjInfo = NULL;

	fc_outerTupDesc = ExecGetResultType(outerPlanState(fc_sortstate));

	/*
	 * 当我们仅对单个按值列进行排序时执行Datum排序，
	 * 否则执行元组排序。
	 */
	if (fc_outerTupDesc->natts == 1 && TupleDescAttr(fc_outerTupDesc, 0)->attbyval)
		fc_sortstate->datumSort = true;
	else
		fc_sortstate->datumSort = false;

	SO1_printf("ExecInitSort: %s\n",
			   "sort node initialized");

	return fc_sortstate;
}

/* ----------------------------------------------------------------
 *		ExecEndSort(node)
 * ----------------------------------------------------------------
 */
void ExecEndSort(SortState *fc_node)
{
	SO1_printf("ExecEndSort: %s\n",
			   "shutting down sort node");

	/*
	 * 清空元组表
	 */
	ExecClearTuple(fc_node->ss.ss_ScanTupleSlot);
	/* 必须删除对排序结果元组的指针 */
	ExecClearTuple(fc_node->ss.ps.ps_ResultTupleSlot);

	/*
	 * 释放tuplesort资源
	 */
	if (fc_node->tuplesortstate != NULL)
		tuplesort_end((Tuplesortstate *) fc_node->tuplesortstate);
	fc_node->tuplesortstate = NULL;

	/*
	 * 关闭子计划
	 */
	ExecEndNode(outerPlanState(fc_node));

	SO1_printf("ExecEndSort: %s\n",
			   "sort node shutdown");
}

/* ----------------------------------------------------------------
 *		ExecSortMarkPos
 *
 *		调用tuplesort以保存当前在排序文件中的位置。
 * ----------------------------------------------------------------
 */
void ExecSortMarkPos(SortState *fc_node)
{
	/*
	 * 如果我们还未排序，则直接返回
	 */
	if (!fc_node->sort_Done)
		return;

	tuplesort_markpos((Tuplesortstate *) fc_node->tuplesortstate);
}


void ExecSortRestrPos(SortState *fc_node)
{
	/*
	 * 如果我们还没有排序，则直接返回。
	 */
	if (!fc_node->sort_Done)
		return;

	/*
	 * 将扫描恢复到之前标记的位置
	 */
	tuplesort_restorepos((Tuplesortstate *) fc_node->tuplesortstate);
}

void ExecReScanSort(SortState *fc_node)
{
	PlanState  *fc_outerPlan = outerPlanState(fc_node);

	/*
	 * 如果我们还没有排序，则直接返回。如果 outerplan 的 chgParam 为
	 * NULL，则不会重新扫描，否则没有理由重新扫描它。
	 */
	if (!fc_node->sort_Done)
		return;

	/* 必须删除对排序结果元组的指针 */
	ExecClearTuple(fc_node->ss.ps.ps_ResultTupleSlot);

	/*
	 * 如果子节点需要重新扫描，则我们忘记之前的排序结果；我们
	 * 必须重新读取子计划并重新排序。如果边界排序参数改变或没有选择
	 * randomAccess，也必须重新排序。
	 *
	 * 否则，我们可以直接倒带并重新扫描已排序的输出。
	 */
	if (fc_outerPlan->chgParam != NULL ||
		fc_node->bounded != fc_node->bounded_Done ||
		fc_node->bound != fc_node->bound_Done ||
		!fc_node->randomAccess)
	{
		fc_node->sort_Done = false;
		tuplesort_end((Tuplesortstate *) fc_node->tuplesortstate);
		fc_node->tuplesortstate = NULL;

		/*
		 * 如果子节点的 chgParam 不为 null，则计划将由
		 * 首先 ExecProcNode 重新扫描。
		 */
		if (fc_outerPlan->chgParam == NULL)
			ExecReScan(fc_outerPlan);
	}
	else
		tuplesort_rescan((Tuplesortstate *) fc_node->tuplesortstate);
}

/* ----------------------------------------------------------------
 *						并行查询支持
 * ----------------------------------------------------------------
 */

/* ----------------------------------------------------------------
 *		ExecSortEstimate
 *
 *		估算传播排序统计信息所需的空间。
 * ----------------------------------------------------------------
 */
void ExecSortEstimate(SortState *fc_node, ParallelContext *fc_pcxt)
{
	Size		fc_size;

	/* 如果没有工具监控或没有工作程序，则不需要此 */
	if (!fc_node->ss.ps.instrument || fc_pcxt->nworkers == 0)
		return;

	fc_size = mul_size(fc_pcxt->nworkers, sizeof(TuplesortInstrumentation));
	fc_size = add_size(fc_size, offsetof(SharedSortInfo, sinstrument));
	shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_size);
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);
}

/* ----------------------------------------------------------------
 *		ExecSortInitializeDSM
 *
 *		为排序统计信息初始化 DSM 空间。
 * ----------------------------------------------------------------
 */
void ExecSortInitializeDSM(SortState *fc_node, ParallelContext *fc_pcxt)
{
	Size		fc_size;

	/* 如果没有工具监控或没有工作程序，则不需要此 */
	if (!fc_node->ss.ps.instrument || fc_pcxt->nworkers == 0)
		return;

	fc_size = offsetof(SharedSortInfo, sinstrument)
		+ fc_pcxt->nworkers * sizeof(TuplesortInstrumentation);
	fc_node->shared_info = shm_toc_allocate(fc_pcxt->toc, fc_size);
	
/* 确保任何未填充的槽将包含零 */
	memset(fc_node->shared_info, 0, fc_size);
	fc_node->shared_info->num_workers = fc_pcxt->nworkers;
	shm_toc_insert(fc_pcxt->toc, fc_node->ss.ps.plan->plan_node_id,
				   fc_node->shared_info);
}

/* ----------------------------------------------------------------
 *		ExecSortInitializeWorker
 *
 *		将工作线程附加到用于排序统计信息的 DSM 空间。
 * ----------------------------------------------------------------
 */
void ExecSortInitializeWorker(SortState *fc_node, ParallelWorkerContext *fc_pwcxt)
{
	fc_node->shared_info =
		shm_toc_lookup(fc_pwcxt->toc, fc_node->ss.ps.plan->plan_node_id, true);
	fc_node->am_worker = true;
}

/* ----------------------------------------------------------------
 *		ExecSortRetrieveInstrumentation
 *
 *		将排序统计信息从 DSM 转移到私有内存。
 * ----------------------------------------------------------------
 */
void ExecSortRetrieveInstrumentation(SortState *fc_node)
{
	Size		fc_size;
	SharedSortInfo *fc_si;

	if (fc_node->shared_info == NULL)
		return;

	fc_size = offsetof(SharedSortInfo, sinstrument)
		+ fc_node->shared_info->num_workers * sizeof(TuplesortInstrumentation);
	fc_si = palloc(fc_size);
	memcpy(fc_si, fc_node->shared_info, fc_size);
	fc_node->shared_info = fc_si;
}
